package com.yuyou.fn.common.database.event;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.github.wens.mybatisplus.toolkit.IdWorker;
import com.yuyou.fn.common.rocketmq.Consumer;
import com.yuyou.fn.common.service.IEmailService;
import com.yuyou.fn.common.util.JSONS;
import com.yuyou.fn.common.util.Threads;
import com.yuyou.fn.common.util.Utils;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.*;


public class HandlerExecutor implements ApplicationContextAware {

    private final static Logger log = LoggerFactory.getLogger(HandlerExecutor.class);

    private ApplicationContext applicationContext;

    private String appName ;

    private String groupName ;

    public String getGroupName() {
        return groupName;
    }

    public void setGroupName(String groupName) {
        this.groupName = groupName;
    }

    public String getAppName() {
        return appName;
    }

    public void setAppName(String appName) {
        this.appName = appName;
    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public void init() {

            Consumer consumer = applicationContext.getBean(Consumer.class);
            Map<String, Object> handlerMap = applicationContext.getBeansWithAnnotation(EventHandler.class);
            if (handlerMap != null) {
                for (Map.Entry<String, Object> entry : handlerMap.entrySet()) {
                    Object service = entry.getValue();
                    if (service instanceof IEventHandler) {
                        IEventHandler eventHandler = (IEventHandler) service;
                        String[] tables = eventHandler.getClass().getAnnotation(EventHandler.class).tables();
                        for (String table : tables) {
                            consumer.subscribe(table.replaceFirst("yyfn\\.",""),String.format("%s-%s", groupName,service.getClass().getSimpleName() ), createMessageListenerOrderly(eventHandler));
                        }
                    }
                }
            }
    }

    private MessageListenerOrderly createMessageListenerOrderly(IEventHandler eventHandler) {
        return (msgs, context)-> {

                    try {
                        for(MessageExt msg : msgs ){
                            JSONObject jsonObject = JSONS.toJSONObject(msg.getBody());
                            String tableName = jsonObject.getString("tableName");
                            String event = jsonObject.getString("event");
                                JSONArray rowList = jsonObject.getJSONArray("rowList");
                                if ("insert".equals(event)) {
                                    try {
                                        long s  = System.currentTimeMillis();
                                        eventHandler.insert(tableName, rowList);
                                        log.info("table : {} , event:insert , handler : {} , times : {} , dataSize:{} " ,tableName,eventHandler.getClass().getName(),System.currentTimeMillis() - s , rowList.size() );
                                    } catch (Exception e) {
                                        logError(tableName,event,eventHandler.getClass().getName(),rowList,e );
                                        throw e ;
                                    }
                                } else if ("update".equals(event)) {
                                    JSONArray updateColumns = jsonObject.getJSONArray("updateColumns");
                                    Set<String> updateColumns2 = new HashSet<String>();
                                    for (int i = 0; i < updateColumns.size(); i++) {
                                        updateColumns2.add(updateColumns.getString(i));
                                    }
                                    try {
                                        long s  = System.currentTimeMillis();
                                        eventHandler.update(tableName, rowList, updateColumns2);
                                        log.info("table : {} , event:update , handler : {} , times : {} , dataSize:{} " ,tableName,eventHandler.getClass().getName(),System.currentTimeMillis() - s , rowList.size() );
                                    } catch (Exception e) {
                                        logError(tableName,event,eventHandler.getClass().getName(),rowList,e );
                                        throw e ;
                                    }
                                } else if ("delete".equals(event)) {
                                    log.info("table : {} event:delete" ,tableName);
                                    try {
                                        long s  = System.currentTimeMillis();
                                        eventHandler.delete(tableName, rowList);
                                        log.info("table : {} , event:delete , handler : {} , times : {} , dataSize:{} " ,tableName,eventHandler.getClass().getName(),System.currentTimeMillis() - s , rowList.size() );
                                    } catch (Exception e) {
                                        logError(tableName,event,eventHandler.getClass().getName(),rowList,e );
                                        throw e ;
                                    }

                                }
                        }
                    }catch (Exception e){
                        log.error("Consume msg fail : msgs="+msgs, e );
                        if( msgs.get(0).getReconsumeTimes() > 5){
                            sendEmail("Try consume " + msgs.get(0).getReconsumeTimes() + " times fail : msg[0]" + msgs.get(0));
                            return ConsumeOrderlyStatus.SUCCESS;
                        }else{
                            context.setSuspendCurrentQueueTimeMillis(3000);
                            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                        }
                    }

                    return ConsumeOrderlyStatus.SUCCESS;
                };
    }

    private void logError(final String tableName , final String event,final String handler ,final JSONArray rowList,Exception e){
        String logTrackingId = String.valueOf(IdWorker.getId());
        log.error("Handler Event fail : logTrackingId="+logTrackingId+",table = "+tableName+",event = " + event + ",handler = " + handler + ",rowList=" + rowList.toJSONString() , e);
        Threads.dispatch(()->sendEmailNotify(  tableName ,  event, handler, logTrackingId));
    }

    private void sendEmailNotify( String tableName , String event,String handler,String logTrackingId ){
        String content = String.format("Server:%s,tableName:%s,event:%s,handler:%s,logTrackingId:%s", Utils.getHostIP(), tableName, event, handler, logTrackingId);
        sendEmail(content);
    }

    private void sendEmail(String content) {
        IEmailService emailService = applicationContext.getBean(IEmailService.class);
        emailService.sendEmail("wenzuojing1@zy.com",appName + "应用告警" ,content,null,null );
    }




}
